package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * Created by Smexy on 2022/11/19
 *
 *      数据分发算子
 *
 *      理论:  在flink中使用一个类  ChannelSelector(通道选择器)，帮你完成上下游算子之间数据的分发！
 *              作用： 等价于  MR 和 SPARK 中的 Partitioner(分区器)
 *
 *                  提供一个方法返回 一条数据 应该发往下游的哪个通道: int selectChannel(T record);
 *
 *      分区器的使用： 无需开发人员自己指定分区器，而是要调用对应的算子，算子会自动绑定不同类型的分区器。
 *                        RescalePartitioner：     rescale     同一个TaskManager负载均衡。
 *                        RebalancePartitioner：   rebalance   全局负载均衡。
 *                                                      默认从source 发往下游 ，使用rebalance。
 *
 *                                                大数据只怕数据倾斜(有的Task忙死，有的Task闲死。闲死的还不能早下班)。
 *
 *                      KeyGroupStreamPartitioner:  keyBy。 最常用。
 *                                                          key相同，发往同一个下游的Task。
 *                                                          key不同，也有可能发往下游的同一个Task。
 *
 *                                                大数据： 从海量数据中提取有价值的信息。
 *                                                          多 -----> 少  聚合。
 *                                                          物以类聚
 *
 *                      GlobalPartitioner      :  global   全局汇总。 把所有数据发往下游的0号通道。
 *
 *                      ForwardPartitioner     : forward. 上下游的并行度必须一致！ 不写分发规则，默认就是forward
 *                                                          帮你1对1 传输
 *                                                              map1  -------> sink1
 *                                                              map2  --------> sink2
 *                                                              map3  --------> sink3
 *                                                       进行Operator chain的必要条件。
 *                                                       Operator chain： 把多个Operator的subTask合并到一个Slot中运行！
 *
 *                       ShufflePartitioner    : shuffle
 *
 *                      CustomPartitionerWrapper  : 自定义单播。
 *                                                  自己编写类，继承这个类。一条数据，只能发往下游的一个通道。
 *              --------------------------------------------------
 *                      BroadcastPartitioner：      broadcast  广播
 *                                                      上游的一条数据，会广播到下游的所有 Task
 *                                                       很少用
 */
public class Demo10_DispathOperator
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888).setParallelism(1);

        ds.map(s -> s).setParallelism(1)//.startNewChain()
          //.broadcast()
          .print().setParallelism(3);

        env.execute();

    }
}
